feat(synapse-core): add piece batcher with message-size addPieces limiter - #933

function createPieceBatcher(
  client: Client<Transport, Chain, Account>,
  options: createPieceBatcher.OptionsType
): PieceBatcher

type createPieceBatcher.OptionsType = {
  dataSet: PdpDataSet | undefined
  wait?: PieceBatcherWait              // default { kind: 'delay', ms: 0 }
  limiter?: Limiter                    // default addPiecesFits
  serviceURL?: string                  // required when dataSet is undefined
  payee?: Address                      // required when dataSet is undefined
  payer?: Address
  metadata?: MetadataObject
  cdn?: boolean
}

type PieceBatcherWait =
  | { kind: 'delay'; ms: number }
  | { kind: 'limiter' }

type PieceBatcher = {
  upload: (input: UploadInput) => Promise<PieceResult>
  pull: (input: PullInput) => Promise<PieceResult>
  enqueue: (piece: EnqueuePiece) => Promise<PieceResult>
  flush: () => Promise<FlushResult | undefined>
  close: () => Promise<void>
  readonly pending: readonly EnqueuePiece[]
  readonly dataSet: PdpDataSet | undefined
}

type UploadInput = {
  data: File | Uint8Array | ReadableStream<Uint8Array>
  size?: number                        // for streams; File uses .size
  metadata?: MetadataObject
  pieceCid?: PieceCID
  onParked?: OnParked
  onProgress?: (bytesUploaded: number) => void
  signal?: AbortSignal
}

type PullInput = {
  pieceCid: PieceCID
  sourceUrl: string
  metadata?: MetadataObject
  onParked?: OnParked
}

type EnqueuePiece = { pieceCid: PieceCID; metadata?: MetadataObject }
type FlushResult = { txHash: Hex; statusUrl: string; pieces: EnqueuePiece[] }
type PieceResult = FlushResult & { pieceCid: PieceCID; batchIndex: number }
type OnParked = (piece: EnqueuePiece) => void | Promise<void>

function addPiecesFits(options: LimiterOptions): boolean
function assertAddPiecesFit(options: LimiterOptions): void
function assertPieceCidSize(pieceCid: PieceCID): void
function estimateAddPiecesCalldataSize(options: LimiterOptions): number

type Limiter = (options: LimiterOptions) => boolean
type LimiterOptions =
  | { kind: 'addPieces'; dataSet?: PdpDataSet; pieces: LimiterPiece[] }
  | { kind: 'createDataSetAndAddPieces'; metadata?: MetadataObject; cdn?: boolean; pieces: LimiterPiece[] }

import { createPieceBatcher } from '@filoz/synapse-core/sp'

const batcher = createPieceBatcher(client, { dataSet })

const [a, b] = await Promise.all([
  batcher.upload({ data: fileA }),
  batcher.upload({ data: fileB }),
])
// a.txHash === b.txHash when they landed in the same window

await batcher.close()

const batcher = createPieceBatcher(client, {
  dataSet: undefined,
  serviceURL: provider.pdp.serviceURL,
  payee: provider.serviceProvider,
  payer: client.account.address,
  cdn: false,
})

await batcher.upload({ data: firstFile })
await batcher.close()
// batcher.dataSet is now the created set; later windows use addPieces

await Promise.all([
  batcher.upload({ data: localFile }),
  batcher.pull({
    pieceCid,
    sourceUrl: `https://primary.example/pdp/piece/${pieceCid}`,
  }),
])
await batcher.close()

import { AddPiecesFlushError } from '@filoz/synapse-core/errors'

try {
  await batcher.upload({
    data: file,
    onParked: ({ pieceCid }) => {
      // piece is on this SP; not yet in an addPieces window
    },
  })
} catch (error) {
  if (AddPiecesFlushError.is(error)) {
    await batcher.enqueue({ pieceCid: error.pieceCid, metadata: error.metadata })
  } else {
    throw error
  }
}

const batcher = createPieceBatcher(client, {
  dataSet,
  wait: { kind: 'limiter' },
})

const p1 = batcher.upload({ data: fileA })
const p2 = batcher.upload({ data: fileB })
await batcher.flush() // or close()
await Promise.all([p1, p2])